package com.think.hadoopTools;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.util.ReflectionUtils;

import java.io.IOException;

public class CompressTool {
    public static void main(String[] args) {
        //以下2个参数不需要带hdfs://
        String sourceFile = null;
        String destFile = null;
        for (int i = 0; i < args.length; i++) {
            if (args[i].equals("-s")) {//要压缩的文件或目录
                sourceFile = args[++i];//示例/user/logsystem/accesslog/19-09-13/
            } else if (args[i].equals("-d")) {//压缩后的文件名
                destFile = args[++i];//示例/user/logsystem/accesslog/19-09-13/1.tar.gz
            }
        }
        System.out.println("sourceFile:" + sourceFile);
        System.out.println("destFile:" + destFile);
        compress(sourceFile, destFile);
    }

    public static void compress(String sourceFile, String destFile) {
        FSDataInputStream in = null;
        CompressionOutputStream out = null;
        //Compressor compressor = null;
        try {
            Class<?> codecClass = Class.forName(GzipCodec.class.getName());
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(conf);
            if (!fs.exists(new Path(sourceFile))) {
                System.out.println("sourceFile不存在：" + sourceFile);
                return;
            }
            CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
            //compressor = CodecPool.getCompressor(codec);
            //指定压缩文件路径
            FSDataOutputStream outputStream = fs.create(new Path(destFile));
            //指定要被压缩的文件路径
            in = fs.open(new Path(sourceFile));
            //创建压缩输出流
            out = codec.createOutputStream(outputStream);
            //out = codec.createOutputStream(outputStream, compressor);
            //IOUtils.copyBytes(in, out, conf);//默认4096byte=4KB
            IOUtils.copyBytes(in, out, 1024 * 1024 * 4);//4M
            out.finish();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(in);
            IOUtils.closeStream(out);
            //如果频繁的使用压缩和解压缩，可以使用CodecPool复用，打开所有compressor的注释
            //并且注释掉IOUtils.closeStream(in);和IOUtils.closeStream(out);因为不需要再关闭流了
            //CodecPool.returnCompressor(compressor);
        }
    }
}
